package com.caius.xiaohashu.comment.biz.consumer;

import com.caius.framework.common.util.JsonUtils;
import com.caius.xiaohashu.comment.biz.constant.MQConstants;
import com.caius.xiaohashu.comment.biz.constant.RedisKeyConstants;
import com.caius.xiaohashu.comment.biz.domain.dataobject.CommentDO;
import com.caius.xiaohashu.comment.biz.domain.mapper.CommentDOMapper;
import com.caius.xiaohashu.comment.biz.model.bo.CommentHeatBO;
import com.caius.xiaohashu.comment.biz.util.HeatCalculator;
import com.github.phantomthief.collection.BufferTrigger;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.scripting.support.ResourceScriptSource;
import org.springframework.stereotype.Component;

import java.math.BigDecimal;
import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * @author Caius
 * @description
 * @since Created in 2025-06-06
 */
@Component
@RocketMQMessageListener(consumerGroup = "xiaohashu_group_" + MQConstants.TOPIC_COMMENT_HEAT_UPDATE,
            topic = MQConstants.TOPIC_COMMENT_HEAT_UPDATE
)
@Slf4j
public class CommentHeatUpdateConsumer implements RocketMQListener<String> {

    @Resource
    private CommentDOMapper commentDOMapper;

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    private BufferTrigger<String> bufferTrigger = BufferTrigger.<String>batchBlocking()
            .bufferSize(50000)
            .batchSize(300)
            .linger(Duration.ofSeconds(2))
            .setConsumerEx(this::consumeMessage)
            .build();

    private void consumeMessage(List<String> bodys) {
        log.info("==> 【评论热度值计算】聚合消息, size: {}", bodys.size());
        log.info("==> 【评论热度值计算】聚合消息, {}", JsonUtils.toJsonString(bodys));

        Set<Long> commentIds =  Sets.newHashSet();
        bodys.forEach(body -> {
            try {
                Set<Long> list = JsonUtils.parseSet(body, Long.class);
                commentIds.addAll(list);
            } catch (Exception e) {
                log.error("", e);
            }
        });
        log.info("==> 去重后的评论 ID: {}", commentIds);

        List<CommentDO> commentDOS = commentDOMapper.selectByCommentIds(commentIds.stream().toList());

        List<Long> ids = Lists.newArrayList();

        List<CommentHeatBO> commentBOS = Lists.newArrayList();

        commentDOS.forEach(commentDO -> {
            Long commentId = commentDO.getId();
            Long likeTotal = commentDO.getLikeTotal();
            Long childCommentTotal = commentDO.getChildCommentTotal();

            BigDecimal heatNum = HeatCalculator.calculateHeat(likeTotal, childCommentTotal);
            ids.add(commentId);
            commentBOS.add(CommentHeatBO.builder()
                    .id(commentId)
                    .heat(heatNum.doubleValue())
                    .noteId(commentDO.getNoteId())
                    .build());
        });

        int count = commentDOMapper.batchUpdateHeatByCommentId(ids, commentBOS);

        if (count == 0) return;

        updateRedisHotComments(commentBOS);
    }

    /**
     * 更新 redis 中间的热点评论
     * @param commentHeatBOList
     */
    private void updateRedisHotComments(List<CommentHeatBO> commentHeatBOList) {
        Map<Long, List<CommentHeatBO>> noteIdAndBOListMap = commentHeatBOList.stream()
                .filter(commentHeatBO -> commentHeatBO.getHeat() > 0)
                .collect(Collectors.groupingBy(CommentHeatBO::getNoteId));

        noteIdAndBOListMap.forEach((noteId, commentHeatBOS) -> {
            String commentListKey = RedisKeyConstants.buildCommentListKey(noteId);

            DefaultRedisScript<Long> script = new DefaultRedisScript<>();

            script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/update_hot_comments.lua")));

            script.setResultType(Long.class);

            List<Object> args = Lists.newArrayList();

            commentHeatBOS.forEach(commentHeatBO -> {
                args.add(commentHeatBO.getId());
                args.add(commentHeatBO.getHeat());
            });

            redisTemplate.execute(script, Collections.singletonList(commentListKey), args.toArray());
        });
    }

    @Override
    public void onMessage(String body) {
        bufferTrigger.enqueue(body);
    }
}
